package drds.server.execute_engine;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.config.Config;
import drds.server.datasource.DataSource;
import drds.server.datasource.GroupDataSourceNode;
import drds.server.datasource.IConnectionHolder;
import drds.server.handler.Handler;
import drds.server.net.packet.ErrorPacket;
import drds.server.route.RouteNode;
import drds.server.server.parser.ServerParse;

/**
 * asyn execute in TaskContext or standalone (TaskContext=null)
 */
public class Task implements Handler, Runnable {

	public static final Logger LOGGER = LoggerFactory.getLogger(Task.class);

	private final String sql;
	private final String dataNodeOrDatabase;
	private IConnectionHolder connectionHolder;
	private final TaskHandler taskHandler;
	private final TaskContext taskContext;
	private final DataSource dataSource;
	private final int id;
	private volatile boolean finished;

	public Task(//
			int id,//
			String sql,//
			String dataNode,//
			TaskHandler taskHandler,//
			TaskContext taskContext) {//
		super();
		this.id = id;
		this.sql = sql;
		this.dataNodeOrDatabase = dataNode;
		this.taskHandler = taskHandler;
		this.taskContext = taskContext;
		this.dataSource = null;
	}

	public Task(//
			String sql, //
			String databaseName,//
			TaskHandler taskHandler,//
			DataSource dataSource) {//
		super();
		this.id = 0;
		this.sql = sql;
		this.dataNodeOrDatabase = databaseName;
		this.taskHandler = taskHandler;
		this.taskContext = null;
		this.dataSource = dataSource;

	}

	public void run() {
		try {
			if (dataSource == null) {
				RouteNode node = new RouteNode(dataNodeOrDatabase, ServerParse.SELECT, sql);
				// create new connection
				Config config = Server.getInstance().getConfig();
				GroupDataSourceNode groupDataSourceNode = config.getGroupDataSourceNodeMap().get(node.getName());
				groupDataSourceNode.getConnection(groupDataSourceNode.getSchemaName(), true, node, this, node);
			} else {
				dataSource.getConnection(dataNodeOrDatabase, true, this, null);
			}
		} catch (Exception e) {
			LOGGER.info("can't get connection for sql ,error:" + e);
			doFinished(true);
		}
	}

	public void teminate(String reason) {
		if (connectionHolder != null) {
			connectionHolder.close(reason);
		}
	}

	@Override
	public void connectionAcquired(final IConnectionHolder connectionHolder) {
		connectionHolder.setHandler(this);
		try {
			connectionHolder.query(sql);
			this.connectionHolder = connectionHolder;
		} catch (Exception e) {// (UnsupportedEncodingException e) {
			doFinished(true);
		}

	}

	public boolean isFinished() {
		return finished;
	}

	private void doFinished(boolean failed) {
		finished = true;
		taskHandler.finished(dataNodeOrDatabase, failed);
		if (taskContext != null) {
			taskContext.onTaskFinished(this);
		}
	}

	@Override
	public void connectionError(IConnectionHolder connectionHolder, Throwable e) {
		LOGGER.info("can't get connection for sql :" + sql);
		doFinished(true);
	}

	@Override
	public void errorResponse(IConnectionHolder connectionHolder, byte[] errorBytes) {
		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.read(errorBytes);

		String errorMsg = "error response errno:" + errorPacket.errorNo + ", " + new String(errorPacket.errorMessage) + " from of sql :" + sql + " at con:" + connectionHolder;

		if (errorPacket.errorNo == 1227) {
			LOGGER.error(errorMsg);

		} else {
			LOGGER.info(errorMsg);
		}

		connectionHolder.release();
		doFinished(true);
	}

	@Override
	public void okResponse(IConnectionHolder connectionHolder, byte[] okBytes) {
		connectionHolder.syncAndExcute();
	}

	@Override
	public void fieldEofResponse(IConnectionHolder connectionHolder, byte[] headerBytes, List<byte[]> fieldBytesList, byte[] eofBytes) {
		taskHandler.onHeader(dataNodeOrDatabase, headerBytes, fieldBytesList);

	}

	@Override
	public void rowResponse(IConnectionHolder connectionHolder, byte[] rowBytes) {
		boolean finsihed = taskHandler.onRowData(dataNodeOrDatabase, rowBytes);
		if (finsihed) {
			connectionHolder.close("not needed by user proc");
			doFinished(false);
		}

	}

	@Override
	public void rowEofResponse(IConnectionHolder connectionHolder, byte[] eofBytes) {
		connectionHolder.release();
		doFinished(false);
	}

	@Override
	public void writeQueueAvailable() {

	}

	@Override
	public void connectionClose(IConnectionHolder connectionHolder, String reason) {
		doFinished(true);
	}

	public int getId() {
		return id;
	}

}
